package main

import (
	"encoding/json"
	"fmt"
	"io/ioutil"
	"net/http"
	"strings"
	"time"

	"github.com/kpango/glg"
)

func WriteToEverylog(tableInfo, data string) error {
	info := strings.Split(tableInfo, "/")
	if len(info) != 2 {
		return fmt.Errorf("kafkaConsume.WriteToEverylog: Bad value of appId/tablename: %s", tableInfo)
	}
	appId := info[0]
	tablename := info[1]
	requests, err := BuildPostRequest(data)
	if err != nil {
		return nil
	}
	for _, req := range requests {
		args, err := BuildWriteArgs(req)
		if err != nil {
			return err
		}
		for _, url := range EverylogUrls(appId, tablename) {
			resp, err := http.Post(
				url,
				"application/x-www-form-urlencoded",
				strings.NewReader(args),
			)
			if err != nil {
				glg.Errorf("kafkaConsume.WriteToEverylog: Failed to connect everylog %s: %v", url, err)
				return err
			}
			defer resp.Body.Close()
			body, _ := ioutil.ReadAll(resp.Body)
			if resp.StatusCode != 200 {
				return fmt.Errorf("kafkaConsume.WriteToEverylog: Everylog %s response: %d, %s", url, resp.StatusCode, string(body))
			}
		}
	}
	return nil
}

type PostRequest struct {
	Data      []Row  `json:"data,omitempty"`
	Dt        string `json:"dt,omitempty"`
	Partition string `json:"partition,omitempty"`
}

type Row struct {
	Key   string            `json:"key,omitempty"`
	Value string            `json:"value,omitempty"`
	Index map[string]string `json:"index,omitempty"`
}

// dt:part:[]Row
type RowsMap map[string]map[string][]Row

func (r *Row) BuildIndex(record map[string]string) {
	r.Index["DATETIME"] = record["DATETIME"]
	r.Index["IP"] = record["IP"]
	r.Index["STAFF_ID"] = record["STAFF_ID"]
	if record["QUERY_TYPE1"] != "" {
		r.Index["QUERY_TYPE1"] = record["QUERY_TYPE1"]
	}
	if record["QUERY_TYPE2"] != "" {
		r.Index["QUERY_TYPE2"] = record["QUERY_TYPE2"]
	}
	if record["QUERY_VALUE1"] != "" {
		r.Index["QUERY_VALUE1"] = record["QUERY_VALUE1"]
	}
	if record["QUERY_VALUE2"] != "" {
		r.Index["QUERY_VALUE2"] = record["QUERY_VALUE2"]
	}
	if record["POST_ID"] != "" {
		r.Index["POST_ID"] = record["POST_ID"]
	}
	if record["ORG_ID"] != "" {
		r.Index["ORG_ID"] = record["ORG_ID"]
	}
}

func BuildPostRequest(data string) ([]PostRequest, error) {
	kvlist := make([]map[string]string, 0)
	data = strings.Replace(data, "'", "\"", -1)
	err := json.Unmarshal([]byte(data), &kvlist)
	if err != nil {
		return nil, err
	}
	rows := make(RowsMap)
	for _, record := range kvlist {
		datetime := record["DATETIME"]
		ip := record["IP"]
		staff := record["STAFF_ID"]
		format := "20060102150405"
		if len(datetime) == 17 {
			format = "20060102150405000"
		}
		timestamp, err := time.Parse(format, datetime)
		if err != nil {
			return nil, err
		}
		dt := timestamp.Format("20060102")
		// partition := timestamp.Format("2006")
		partition := record["STAFF_ID"][0:1]
		record_json, err := json.Marshal(record)
		if err != nil {
			return nil, err
		}
		row := Row{
			// prefix "_crmlog_" for query all
			Key:   fmt.Sprintf("_crmlog_%s|%s|%s", staff, datetime, ip),
			Value: string(record_json),
			Index: make(map[string]string),
		}
		row.BuildIndex(record)
		if rows[dt] == nil {
			rows[dt] = make(map[string][]Row)
		}
		if rows[dt][partition] == nil {
			rows[dt][partition] = make([]Row, 0)
		}
		rows[dt][partition] = append(rows[dt][partition], row)
	}
	var args []PostRequest
	for dt, _ := range rows {
		for partition, _ := range rows[dt] {
			args = append(args, PostRequest{
				Data:      rows[dt][partition],
				Dt:        dt,
				Partition: partition,
			})
		}
	}
	return args, nil
}

func BuildWriteArgs(req PostRequest) (string, error) {
	data, err := json.Marshal(req.Data)
	if err != nil {
		return "", err
	}
	res := fmt.Sprintf("dt=%s&partition=%s&data=%s", req.Dt, req.Partition, string(data))
	return res, nil
}

func EverylogUrls(appid, tablename string) []string {
	var res []string
	urls := strings.Split(*EverlogUrl, ",")
	for _, url := range urls {
		url = strings.TrimSpace(url)
		res = append(res, fmt.Sprintf("http://%s/everylog/%s/%s", url, appid, tablename))
	}
	return res
}

